package cn.itcast.szse;

import cn.itcast.avro.SzseAvro;
import cn.itcast.kafka.KafkaPro;
import cn.itcast.util.DateTimeUtil;
import cn.itcast.util.DateUtil;
import org.apache.kafka.clients.producer.KafkaProducer;

import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.net.Socket;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;

/**
 * 深市客户端代码：主要接受服务端对外广播的行情数据
 */
public class SocketClient {
    //随机浮动成交价格系数
    private static Double[] price = new Double[]{0.1, 0.11, 0.12, 0.13, 0.14, 0.15, 0.16, 0.17, 0.18, 0.19, 0.2, -0.1, -0.11, -0.12, -0.13, -0.14, -0.15, -0.16, -0.17, -0.18, -0.19, -0.2};

    //随机浮动成交量
    private static int[] volumn = new int[]{50, 80, 110, 140, 170, 200, 230, 260, 290, 320, 350, 380, 400, 500, 600, 700, 800, 900, 1000, 1100, 1200, 1300};

    //
    private static Map<String,Map<String,Long>> map = new HashMap<>();

    public static void main(String[] args) throws IOException {
        /**
         * 1.建立socket链接
         * 2.读文件缓存成交量和成交金额
         * 3.解析行数据，数据转换
         * 4.发送kafka
         */

        //todo 1.建立socket链接
        Socket socket = new Socket("localhost", 4444);
        //todo 2.读文件缓存成交量和成交金额
        DataInputStream dataInputStream = new DataInputStream(socket.getInputStream());
        //todo 3.解析行数据，数据转换
        KafkaPro kafkaPro = new KafkaPro();
        while (true){
            //读取每一条数据
            String str = dataInputStream.readUTF();
            //3.解析行数据，数据转换
            SzseAvro szseAvro = parseStr(str);
            //todo 4.发送kafka
            kafkaPro.sendData("szse",szseAvro);
        }
    }

    //解析字符串
    private static SzseAvro parseStr(String str) {
        /**
         * 1.拷贝成交量和成交价数组
         * 2.获取随机浮动成交量和价格
         * 3.字符串切割，计算最新价
         * 4.获取缓存的成交量/金额
         * 5.计算总成交量/金额
         * 6.缓存总成交量/金额
         * 7.获取最高价和最低价（和最新价比较）
         * 8.分装结果数据
         */

        //2.获取随机浮动成交量和价格
        Random random = new Random();
        int vIndex = random.nextInt(volumn.length);
        int pIndex = random.nextInt(price.length);
        int randomVolumn = volumn[vIndex];  //随机浮动成交量
        Double randomPrice = price[pIndex]; //随机浮动价格系数

        //3.字符串切割，计算最新价
        String[] arr = str.split("\\|");
        BigDecimal tradePrice = new BigDecimal(arr[9].trim()); //最新价，金额都用bigdecimal
        //26.43*(1+0.11)  保留两位小数点，四舍五入 ,最新价覆盖老的价格
        tradePrice = tradePrice.multiply(new BigDecimal(1 + randomPrice)).setScale(2, RoundingMode.HALF_UP);

        //文件中成交量
        long tradeVol = Long.parseLong(arr[3].trim());
        //文件中成交金额,增量成交金额
        long tradeAmt = Double.valueOf(arr[4].trim()).longValue();


        //计算总成交量和总成交金额
        Long totalVol = 0l;
        Long totalAmt = 0l;
        String code = arr[1].trim(); //产品代码
        Map<String, Long> amtVolMap = map.get(code);
        if(amtVolMap==null){
            totalVol = tradeVol;
            totalAmt = tradeAmt;
            //6.缓存总成交量和总成交金额
            HashMap<String, Long> mapCache = new HashMap<>();
            mapCache.put("tradeVol",totalVol);
            mapCache.put("tradeAmt",totalAmt);
            map.put(code,mapCache);
        }else {
            //4.获取缓存的成交量/金额
            Long tradeVolTmp = amtVolMap.get("tradeVol");
            Long tradeAmtTmp = amtVolMap.get("tradeAmt");
            totalVol = tradeVolTmp+randomVolumn; //总成交量
            //总成交金额 != tradePrice+tradeVol
            //增量的成交额 = 浮动成交量*最新价
            //总成交额=当前总金额+增量的成交金额；
            BigDecimal tmpAmt = tradePrice.multiply(new BigDecimal(randomVolumn));
            totalAmt = tradeAmtTmp + tmpAmt.longValue();

            //6.缓存总成交量和总成交金额
            HashMap<String, Long> mapCache = new HashMap<>();
            mapCache.put("tradeVol",totalVol);
            mapCache.put("tradeAmt",totalAmt);
            map.put(code,mapCache);
        }

        //7.获取最高价和最低价（和最新价比较）
        BigDecimal highPrice = new BigDecimal(arr[7].trim());
        BigDecimal lowPrice = new BigDecimal(arr[8].trim());

        if(tradePrice.compareTo(highPrice)==1){ //>
            highPrice=tradePrice;
        }
        if(tradePrice.compareTo(lowPrice)==-1){ //<
            lowPrice=tradePrice;
        }

        //封装结果数据
        SzseAvro szseAvro = new SzseAvro();

        szseAvro.setMdStreamID(arr[0].trim());
        szseAvro.setSecurityID(arr[1].trim());
        szseAvro.setSymbol(arr[2].trim());
        szseAvro.setTradeVolume(totalVol); //成交数量
        szseAvro.setTotalValueTraded(totalAmt); //总成交金额
        szseAvro.setPreClosePx(Double.parseDouble(arr[5].trim()));//昨日收盘价
        szseAvro.setOpenPrice(Double.parseDouble(arr[6].trim()));//今日开盘价
        szseAvro.setHighPrice(highPrice.doubleValue()); //最高价
        szseAvro.setLowPrice(lowPrice.doubleValue());//最低价
        szseAvro.setTradePrice(tradePrice.doubleValue());//最新价
        szseAvro.setClosePx(tradePrice.doubleValue());//今日收盘价,在15点之前，收盘价和最新价数据一样
        szseAvro.setTradingPhaseCode("T11");//处于连续竞价期间，市场交易未结束
        szseAvro.setTimestamp(new Date().getTime());
        return szseAvro;
    }
}
